  1.Kafka高级特性解析之物理存储中日志存储下的索引
   
   偏移量索引文件用于记录消息偏移量与物理地址之间的映射关系。时间戳索引文件则根据时间戳查找对应的偏移量。
   文件
   查看一个topic分区目录下的内容，发现有log、index和timeindex三个文件：
   (1).log文件名是以文件中第一条message的offset来命名的，实际offset长度是64位，但是这里只使用了20位，应付
生产是足够的。
   (2).一组index+log+timeindex文件的名字是一样的，并且log⽂件默认写满1G后，会进行log rolling形成一个新的组
合来记录消息，这个是通过broker端log.segment.bytes = 1073741824 指定的。
   (3).index和timeindex在刚使用时会分配10M的大小，当进行log rolling 后，它会修剪为实际的大小。
   创建主题
   [root@linux121 ~]# kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic
tp_demo_05 --partitions 1 --replication-factor 1 --config segment.bytes=104857600
   创建消息文件
   [root@linux121 ~]# for i in `seq 10000000`; do echo "hello lagou $i" >> nmm.txt; done
   [root@linux121 ~] ll -h nmm.txt
   将文本消息生产到主题中
   [root@linux121 ~]# kafka-console-producer.sh --broker-list linux121:9092 --topic tp_demo_05
< nmm.txt
   查看存储文件
   [root@linux121 ~] cd /var/lagou/kafka/kafka-logs/tp_demo_05-0
   [root@linux121 tp_demo_05-0]# ll
   [root@linux121 tp_demo_05-0]# pwd
   /var/lagou/kafka/kafka-logs/tp_demo_05-0
   如果想查看这些文件，可以使用kafka提供的shell来完成，几个关键信息如下：
   (1).offset是逐渐增加的整数，每个offset对应一个消息的偏移量。
   (2).position：消息批字节数，用于计算物理地址。
   (3).CreateTime：时间戳。
   (4).magic：2代表这个消息类型是V2，如果是0则代表是V0类型，1代表V1类型。
   (5).compresscodec：None说明没有指定压缩类型，kafka目前提供了4种可选择，0-None、1-GZIP、2-snappy、3-lz4。
   (6).crc：对所有字段进行校验后的crc值。
   [root@linux121 tp_demo_05-0]# kafka-run-class.sh kafka.tools.DumpLogSegments --files
00000000000000000000.log --print-data-log | head
Dumping 00000000000000000000.log
Starting offset: 0
baseOffset: 0 lastOffset: 716 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 0
CreateTime: 1596513421661 isvalid: true size: 16380 magic: 2 compresscodec: NONE crc:
2973274901
baseOffset: 717 lastOffset: 1410 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 16380
CreateTime: 1596513421715 isvalid: true size: 16371 magic: 2 compresscodec: NONE crc:
1439993110
baseOffset: 1411 lastOffset: 2092 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 32751
CreateTime: 1596513421747 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc:
3528903590
baseOffset: 2093 lastOffset: 2774 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 49116
CreateTime: 1596513421791 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc:
763876977
baseOffset: 2775 lastOffset: 3456 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 65481
CreateTime: 1596513421795 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc:
2218198476
baseOffset: 3457 lastOffset: 4138 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 81846
CreateTime: 1596513421798 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc:
4018065070
baseOffset: 4139 lastOffset: 4820 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 98211
CreateTime: 1596513421802 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc:
3073882858
baseOffset: 4821 lastOffset: 5502 baseSequence: -1 lastSequence: -1 producerId: -1
producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 114576
CreateTime: 1596513421819 isvalid: true size: 16365 magic: 2 compresscodec: NONE crc:
207330377
  [root@linux121 tp_demo_05-0]#
  关于消息偏移量
  1).消息存储
  (1).消息内容保存在log日志文件中。
  (2).消息封装为Record，追加到log日志文件末尾，采用的是顺序写模式。
  (3).一个topic的不同分区，可认为是queue，顺序写入接收到的消息。
  消费者有offset。下图中，消费者A消费的offset是9，消费者B消费的offset是11，不同的消费者offset是交给一个
内部公共topic来记录的。
  时间戳索引文件，它的作用是可以让用户查询某个时间段内的消息，它一条数据的结构是时间戳(8byte)+相对offset 
(4byte)，如果要使用这个索引文件，首先需要通过时间范围，找到对应的相对offset，然后再去对应的index文件找
到position信息，然后才能遍历log文件，它也是需要使用上面说的index文件的。
  但是由于producer生产消息可以指定消息的时间戳，这可能将导致消息的时间戳不一定有先后顺序，因此尽量不要
生产消息时指定时间戳。
  2).偏移量
   (1).位置索引保存在index文件中
   (2).log日志默认每写入4K（log.index.interval.bytes设定的），会写入一条索引信息到index文件中，因此索引
文件是稀疏索引，它不会为每条日志都建立索引信息。
   (3).log日件中的日志，是顺序写⼊的，由message+实际offset+position组成
   (4).索引文件的数据结构则是由相对offset（4byte）+position（4byte）组成，由于保存的是相对第一个消息的
相对offset，只需要4byte就可以了，可以节省空间，在实际查找后还需要计算回实际的offset，这对用户是透明的。
  稀疏索引，索引密度不高，但是offset有序，二分查找的时间复杂度为O(lgN)，如果从头遍历时间复杂度是O(N)。
  偏移量索引由相对偏移量和物理地址组成。
  可以通过如下命令解析.index 文件
  kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000000.index --print-data-log
 | head
  注意：offset与position没有直接关系，因为会删除数据和清理日志
  [root@node1 tp_demo_05-0]# kafka-run-class.sh kafka.tools.DumpLogSegments --files
00000000000003925423.log --print-data-log | head
Dumping 00000000000003925423.log
Starting offset: 3925423
baseOffset: 3925423 lastOffset: 3926028 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 0
CreateTime: 1596513434779 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
4049330741
baseOffset: 3926029 lastOffset: 3926634 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 16359
CreateTime: 1596513434786 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
2290699169
baseOffset: 3926635 lastOffset: 3927240 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 32718
CreateTime: 1596513434787 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
368995405
baseOffset: 3927241 lastOffset: 3927846 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 49077
CreateTime: 1596513434788 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
143415655
baseOffset: 3927847 lastOffset: 3928452 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 65436
CreateTime: 1596513434789 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
572340120
baseOffset: 3928453 lastOffset: 3929058 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 81795
CreateTime: 1596513434790 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
1029643347
baseOffset: 3929059 lastOffset: 3929664 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 98154
CreateTime: 1596513434791 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
2163818250
baseOffset: 3929665 lastOffset: 3930270 baseSequence: -1 lastSequence: -1 producerId:
-1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 114513
CreateTime: 1596513434792 isvalid: true size: 16359 magic: 2 compresscodec: NONE crc:
3747213735
  [root@node1 tp_demo_05-0]#
  在偏移量索引文件中,索引数据都是顺序记录offset,但时间戳索引文件中每个追加的索引时间戳必须大于之前
追加的索引项,否则不予追加。在Kafka0.11.0.0 以后,消息元数据中存在若干的时间戳信息。如果 broker端参
数 log.message.timestamp.type 设置为 LogAppendTIme ，那么时间戳必定能保持单调增长。反之如果是
CreateTime 则无法保证顺序。
  注意：timestamp文件中的offset与index文件中的relativeOffset不是一一对应的。因为数据的写入是各自追
加。
  思考：如何查看偏移量为23的消息？
  Kafka 中存在⼀个 ConcurrentSkipListMap 来保存在每个日志分段,通过跳跃表方式，定位到在
00000000000000000000.index ，通过⼆分法在偏移量索引文件中找到不大于23的最大索引项，即offset20那栏，
然后从日志分段文件中的物理位置为320 开始顺序查找偏移量为23的消息。
  3).时间戳
  在偏移量索引文件中,索引数据都是顺序记录offset,但时间戳索引文件中每个追加的索引时间戳必须大于之前
追加的索引项,否则不予追加。在 Kafka 0.11.0.0 以后,消息信息中存在若⼲的时间戳信息。如果broker端参数
log.message.timestamp.type 设置为 LogAppendTIme ，那么时间戳必定能保持单调增长。反之如果是CreateTime
则无法保证顺序。
  通过时间戳方式进行查找消息，需要通过查找时间戳索引和偏移量索引两个文件。
  时间戳索引索引格式：前八个字节表示时间戳，后四个字节表示偏移量。
  思考：查找时间戳为1557554753430开始的消息？
  (1).查找该时间戳应该在哪个日志分段中。将1557554753430和每个日志分段中最大时间戳largestTimeStamp
逐一对比，直到找到不⼩于1557554753430所对应的日志分段。日志分段中的largestTimeStamp的计算是：先查
询该日志分段所对应时间戳索引文件,找到最后一条索引项,若最后一条索引项的时间戳字段值大于0，则取该值，
否则取该日志分段的最近修改时间。
  (2).查找该日志分段的偏移量索引文件，查找该偏移量对应的物理地址。
  (3).日志文件中从320的物理位置开始查找不小于1557554753430数据。
  注意:timestamp文件中的offset与index文件中的relativeOffset不是一一对应的,因为数据的写入是各自追
加。
  2.清理
  
  Kafka 提供两种日志清理策略：
  日志删除：按照一定的删除策略，将不满足条件的数据进行数据删除
  日志压缩：针对每个消息的Key进行整合,对于有相同Key的不同Value值,只保留最后一个版本。
  Kafka 提供 log.cleanup.policy 参数进行相应配置，默认值:delete,还可以选择compact 。
  主题级别的配置项是 cleanup.policy 。
  1).日志删除之基于时间
  日志删除任务会根据 log.retention.hours/log.retention.minutes/log.retention.ms 设定日志保留的时
间节点。如果超过该设定值，就需要进行删除。默认是 7 天， log.retention.ms 优先级最高。
  Kafka 依据日志分段中最大的时间戳进行定位。
  首先要查询该日志分段所对应的时间戳索引文件,查找时间戳索引文件中最后一条索引项,若最后一条索引项的
时间戳字段值大于0，则取该值，否则取最近修改时间。
  为什么不直接选最近修改时间呢？
  因为日志文件可以有意无意的被修改，并不能真实的反应日志分段的最大时间信息。
  删除过程
  (1).从日志对象中所维护日志分段的跳跃表中移除待删除的日志分段,保证没有线程对这些日志分段进行读取操
作
  (2).这些日志分段所有文件添加 上 .delete 后缀。
  (3).交由一个以 "delete-file" 命名的延迟任务来删除这些 .delete 为后缀的文件。延迟执行时间可以通过
file.delete.delay.ms 进行设置
  如果活跃的日志分段中也存在需要删除的数据时？
  Kafka 会先切分出一个新的日志分段作为活跃日志分段，该日志分段不删除，删除原来的日志分段。
  先腾出地方，再删除。
  2).日志删除之基于日志大小
  日志删除任务会检查当前日志的大小是否超过设定值。设定项为 log.retention.bytes,单个日志分段的大小
由 log.segment.bytes 进行设定。
  删除过程
  (1).计算需要被删除的日志总大小 (当前日志文件大小（所有分段）减去retention值)。
  (2).从日志文件第一个 LogSegment 开始查找可删除的日志分段的文件集合。
  (3).执行删除
  3).日志删除之基于偏移量
  根据日志分段的下一个日志分段的起始偏移量是否大于等于日志文件的起始偏移量,若是,则可以删除此日志分
段。
  注意:日志文件的起始偏移量并不一定等于第一个日志分段的基准偏移量,存在数据删除,可能与之相等的那条
数据已经被删除了
  删除过程
  (1).从头开始遍历每个日志分段,日志分段1的下一个日志分段的起始偏移量为21,小于logStartOffset,将日志
分段1加入到删除队列中
  (2).日志分段2的下一个日志分段的起始偏移量为35,小于logStartOffset,将日志分段2加入到删除队列中
  (3).日志分段3的下一个日志分段的起始偏移量为57,小于logStartOffset,将⽇志分段3加人删除集合中
  (4).日志分段4的下一个日志分段的其实偏移量为71,小于logStartOffset，则不进行删除。
  4).日志压缩策略
  概念
  日志压缩是Kafka的一种机制，可以提供较为细粒度的记录保留，而不是基于粗粒度的基于时间的保留。
  对于具有相同的Key，而数据不同，只保留最后一条数据，前面的数据在合适的情况下删除。
  应用场景
  日志压缩特性,就实时计算来说,可以在异常容灾方面有很好的应用途径。比如,我们在Spark、Flink中做实时
计算时,需要长期在内存里面维护一些数据,这些数据可能是通过聚合了一天或者一周的日志得到的,这些数据一旦
由于异常因素(内存、网络、磁盘等)崩溃了,从头开始计算需要很长的时间.一个比较有效可行的方式就是定时将
内存里的数据备份到外部存储介质中，当崩溃出现时，再从外部存储介质中恢复并继续计算。
  使用日志压缩来替代这些外部存储有哪些优势及好处呢？这里为大家列举并总结了几点：
  Kafka即是数据源又是存储工具，可以简化技术栈，降低维护成本
  使用外部存储介质的话，需要将存储的Key记录下来，恢复的时候再使用这些Key将数据取回，实现起来有一
定的工程难度和复杂度。使用Kafka的日志压缩特性，只需要把数据写进Kafka，等异常出现恢复任务时再读回
到内存就可以了
  Kafka对于磁盘的读写做了大量的优化工作,比如磁盘顺序读写.相对于外部存储介质没有索引查询等工作量
的负担，可以实现高性能。同时，Kafka的日志压缩机制可以充分利用廉价的磁盘，不用依赖昂贵的内存来处
理，在性能相似的情况下，实现非常高的性价比（这个观点仅仅针对于异常处理和容灾的场景来说）
  5).日志压缩方式的实现细节
  主题的cleanup.policy 需要设置为compact。
  Kafka的后台线程会定时将Topic遍历两次：
  (1).记录每个key的hash值最后一次出现的偏移量
  (2).第二次检查每个offset对应的Key是否在后面的日志中出现过，如果出现了就删除对应的日志。
  日志压缩允许删除，除最后一个key之外，删除先前出现的所有该key对应的记录。在一段时间后从日志中清理，
以释放空间。
  注意：日志压缩与key有关，确保每个消息的key不为null。
  压缩是在Kafka后台通过定时重新打开Segment来完成的，Segment的压缩细节如下图所示：
  日志压缩可以确保：
  (1).任何保持在日志头部以内的使用者都将看到所写的每条消息，这些消息将具有顺序偏移量。可以使用Topic
的min.compaction.lag.ms属性来保证消息在被压缩之前必须经过的最短时间。也就是说，它为每个消息在(未压缩)
头部停留的时间提供了一个下限。可以使⽤Topic的max.compaction.lag.ms属性来保证从收到消息到消息符合压缩
条件之间的最大延时
  消息始终保持顺序，压缩永远不会重新排序消息，只是删除一些而已
  消息的偏移量永远不会改变，它是日志中位置的永久标识符
  从日志开始的任何使用者将至少看到所有记录的最终状态，按记录的顺序写入。另外，如果使用者在比Topic的
log.cleaner.delete.retention.ms短的时间内到达日志的头部，则会看到已删除记录的所有delete标记。保留时
间默认是24小时。
  默认情况下，启动日志清理器，若需要启动特定Topic的日志清理,请添加特定的属性。配置日志清理器,这里为
大家总结了以下几点：
  (1).log.cleanup.policy 设置为compact ，Broker的配置，影响集群中所有的Topic。
  (2).log.cleaner.min.compaction.lag.ms ，用防止对更新超过最小消息进行压缩，如果没有设置，除最后一
个Segment之外，所有Segment都有资格进行压缩
  log.cleaner.max.compaction.lag.ms ，用于防止低生产速率的日志在无限制的时间内不压缩。
  Kafka的日志压缩原理并不复杂，就是定时把所有的日志读取两遍，写一遍，而CPU的速度超过磁盘完全不是问
题，只要日志的量对应的读取两遍和写入一遍的时间在可接受的范围内，那么它的性能就是可以接受的。
   
  